# Copyright (c) HySoP 2011-2024
#
# This file is part of HySoP software.
# See "https://particle_methods.gricad-pages.univ-grenoble-alpes.fr/hysop-doc/"
# for further info.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
@file advection.py
RestrictionFilter operator generator.
"""
from hysop.constants import Implementation
from hysop.tools.htypes import check_instance, to_list, first_not_None
from hysop.tools.decorators import debug
from hysop.tools.henum import EnumFactory
from hysop.fields.continuous_field import Field, ScalarField
from hysop.topology.cartesian_descriptor import CartesianTopologyDescriptors
from hysop.parameters.scalar_parameter import ScalarParameter
from hysop.core.graph.computational_node import ComputationalGraphNode
from hysop.core.graph.node_generator import ComputationalGraphNodeGenerator
from hysop.core.graph.computational_node_frontend import (
MultiComputationalGraphNodeFrontend,
)
from hysop.testsenv import __HAS_OPENCL_BACKEND__
FilteringMethod = EnumFactory.create(
"FilteringMethod", ["SPECTRAL", "REMESH", "POLYNOMIAL", "SUBGRID"]
)
[docs]
class SpatialFilterFrontend(MultiComputationalGraphNodeFrontend):
def __new__(
cls,
input_variable,
output_variable,
filtering_method,
implementation=None,
base_kwds=None,
**kwds,
):
return super().__new__(
cls,
input_field=None,
input_topo=None,
output_field=None,
output_topo=None,
implementation_key=None,
implementation=None,
base_kwds=base_kwds,
**kwds,
)
def __init__(
self,
input_variable,
output_variable,
filtering_method,
implementation=None,
base_kwds=None,
**kwds,
):
"""
Initialize a SpatialFilter operator.
Parameters
----------
input_variable: ScalarField
Input field as a tuple (ScalarField, CartesianTopologyDescriptor).
output_variable: ScalarField
Output field as a tuple (ScalarField, CartesianTopologyDescriptor).
filtering_method: FilteringMethod
Specify the lowpass filter type (either spectral or with remeshing kernels).
implementation: implementation, optional, defaults to None
target implementation, should be contained in available_implementations().
If None, implementation will be set to default_implementation().
base_kwds: dict, optional, defaults to None
Base class keywords arguments.
If None, an empty dict will be passed.
kwds:
Extra parameters passed to generated operator.
Notes
-----
An implementation should at least support the SpatialFilterBase interface.
"""
check_instance(input_variable, tuple, size=2)
check_instance(output_variable, tuple, size=2)
check_instance(filtering_method, FilteringMethod)
input_field, input_topo = input_variable
output_field, output_topo = output_variable
check_instance(input_field, ScalarField)
check_instance(output_field, ScalarField)
check_instance(input_topo, CartesianTopologyDescriptors)
check_instance(output_topo, CartesianTopologyDescriptors)
check_instance(base_kwds, dict, keys=str, allow_none=True)
assert input_topo != output_topo, "Same topology for input and output."
super().__init__(
input_field=input_field,
input_topo=input_topo,
output_field=output_field,
output_topo=output_topo,
implementation_key=filtering_method,
implementation=implementation,
base_kwds=base_kwds,
**kwds,
)
[docs]
class RestrictionFilterFrontend(SpatialFilterFrontend):
"""
Interface for restriction filtering: fine grid -> coarse grid
Available implementations are:
*Python/OpenCL using spectral filtering
*Python using remeshing kernels
*Python by just taking a subgrid (compatibility with deprecated MultiresolutionFilter)
"""
[docs]
@classmethod
def all_implementations(cls):
from hysop.backend.host.python.operator.spatial_filtering import (
PythonRemeshRestrictionFilter,
PythonSpectralRestrictionFilter,
PythonSubgridRestrictionFilter,
PythonPolynomialRestrictionFilter,
)
ai = {
FilteringMethod.SUBGRID: {
Implementation.PYTHON: PythonSubgridRestrictionFilter,
},
FilteringMethod.POLYNOMIAL: {
Implementation.PYTHON: PythonPolynomialRestrictionFilter,
},
FilteringMethod.SPECTRAL: {
Implementation.PYTHON: PythonSpectralRestrictionFilter,
},
FilteringMethod.REMESH: {
Implementation.PYTHON: PythonRemeshRestrictionFilter,
},
}
if __HAS_OPENCL_BACKEND__:
from hysop.backend.device.opencl.operator.spatial_filtering import (
OpenClSpectralRestrictionFilter,
OpenClSubgridRestrictionFilter,
OpenClPolynomialRestrictionFilter,
)
ai[FilteringMethod.SUBGRID].update(
{
Implementation.OPENCL: OpenClSubgridRestrictionFilter,
}
)
ai[FilteringMethod.POLYNOMIAL].update(
{
Implementation.OPENCL: OpenClPolynomialRestrictionFilter,
}
)
ai[FilteringMethod.SPECTRAL].update(
{
Implementation.OPENCL: OpenClSpectralRestrictionFilter,
}
)
return ai
[docs]
@classmethod
def all_default_implementations(cls):
adi = {
FilteringMethod.SUBGRID: Implementation.PYTHON,
FilteringMethod.POLYNOMIAL: Implementation.PYTHON,
FilteringMethod.SPECTRAL: Implementation.PYTHON,
FilteringMethod.REMESH: Implementation.PYTHON,
}
return adi
[docs]
class InterpolationFilterFrontend(SpatialFilterFrontend):
"""
Interface for interpolation filtering: coarse grid -> fine grid
Available implementations are:
*Python using polynomials (linear, cubic, quintic, ...)
"""
[docs]
@classmethod
def all_implementations(cls):
from hysop.backend.host.python.operator.spatial_filtering import (
PythonPolynomialInterpolationFilter,
)
from hysop.backend.device.opencl.operator.spatial_filtering import (
OpenClPolynomialInterpolationFilter,
)
ai = {
FilteringMethod.POLYNOMIAL: {
Implementation.PYTHON: PythonPolynomialInterpolationFilter,
Implementation.OPENCL: OpenClPolynomialInterpolationFilter,
},
}
return ai
[docs]
@classmethod
def all_default_implementations(cls):
adi = {
FilteringMethod.POLYNOMIAL: Implementation.PYTHON,
}
return adi
[docs]
class SpatialFilter(ComputationalGraphNodeGenerator):
"""
Graphnode generator to generate interpolation or restriction filter for multiple fields at once.
"""
@debug
def __new__(
cls,
input_variables,
output_variables,
filtering_method,
implementation=None,
base_kwds=None,
**kwds,
):
base_kwds = first_not_None(base_kwds, {})
return super().__new__(
cls,
candidate_input_tensors=None,
candidate_output_tensors=None,
**base_kwds,
)
@debug
def __init__(
self,
input_variables,
output_variables,
filtering_method,
implementation=None,
base_kwds=None,
**kwds,
):
"""
Initialize a RestrictionFilter/InterpolationFilter operator generator.
Parameters
----------
intput_variables: dict
Input fields on fine grid.
Dictionary of continuous fields as keys and topologies as values.
output_variables: dict
Output fields on coarse grid, default to input_fields.
Dictionary of continuous fields as keys and topologies as values.
filtering_method: FilteringMethod
Specify the lowpass filter type (either spectral or with remeshing kernels).
implementation: implementation, optional, defaults to None
target implementation, should be contained in available_implementations().
If None, implementation will be set to default_implementation().
base_kwds: dict, optional, defaults to None
Base class keywords arguments.
If None, an empty dict will be passed.
kwds:
Extra parameters passed to generated operators.
"""
input_fields = list(
ComputationalGraphNode.expand_tensor_fields(input_variables.keys())[0]
)
assert len(set(input_fields)) == len(input_fields)
output_fields = list(
ComputationalGraphNode.expand_tensor_fields(output_variables.keys())[0]
)
assert len(input_fields) == len(output_fields)
for i, field in enumerate(output_fields):
if field is None:
output_fields[i] = input_fields[i]
input_fields = tuple(input_fields)
output_fields = tuple(output_fields)
base_kwds = first_not_None(base_kwds, {})
check_instance(input_fields, tuple, values=ScalarField)
check_instance(output_fields, tuple, values=ScalarField)
check_instance(
input_variables, dict, keys=Field, values=CartesianTopologyDescriptors
)
check_instance(
output_variables, dict, keys=Field, values=CartesianTopologyDescriptors
)
check_instance(base_kwds, dict, keys=str)
check_instance(filtering_method, FilteringMethod)
check_instance(implementation, Implementation)
super().__init__(
candidate_input_tensors=None, candidate_output_tensors=None, **base_kwds
)
self._input_fields = input_fields
self._output_fields = output_fields
self._input_variables = input_variables
self._output_variables = output_variables
self._impl = implementation
self._fm = filtering_method
self._kwds = kwds
@debug
def _generate(self):
nodes = []
for ifield, ofield in zip(self._input_fields, self._output_fields):
stopo = ComputationalGraphNode.get_topo_discretization(
self._input_variables, ifield
)
ttopo = ComputationalGraphNode.get_topo_discretization(
self._output_variables, ofield
)
check_instance(stopo, tuple, values=int)
check_instance(ttopo, tuple, values=int)
assert len(stopo) == len(ttopo)
fm = self._fm
impl = self._impl
kwds = self._kwds.copy()
# if source topology is destination topology there is nothing to be done
if ttopo == stopo:
continue
elif all(ns <= nt for (ns, nt) in zip(stopo, ttopo)):
# here we build an interpolation filter operator
node = InterpolationFilterFrontend(
input_variable=(ifield, stopo),
output_variable=(ofield, ttopo),
filtering_method=fm,
implementation=impl,
**kwds,
)
elif all(ns >= nt for (ns, nt) in zip(stopo, ttopo)):
# here we build a restriction filter operator
node = RestrictionFilterFrontend(
input_variable=(ifield, stopo),
output_variable=(ofield, ttopo),
filtering_method=fm,
implementation=impl,
**kwds,
)
else:
msg = (
"Inconsistant topology descriptors {} and {} for field {} and {}, "
)
msg += "cannot interpolate and restrict at the same time."
msg = msg.format(stopo, ttopo, ifield.name, ofield.name)
raise RuntimeError(msg)
nodes.append(node)
return nodes